package com.atguigu.hightest;

import com.atguigu.bean.HotItem;
import com.atguigu.bean.UserBehavior;
import com.atguigu.utils.ITtoListUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Comparator;
import java.util.List;

/**
 * 需求：每隔5分钟输出最近1小时内点击量最多的前N个商品
 */
public class Flink04_Project_Product_TopN {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);

        DataStreamSource<String> ds = env.readTextFile("input/UserBehavior.csv");

        ds.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String value) throws Exception {
                String[] datas = value.split(",");
                return new UserBehavior(
                        Long.valueOf(datas[0]),
                        Long.valueOf(datas[1]),
                        Integer.valueOf(datas[2]),
                        datas[3],
                        Long.valueOf(datas[4]) * 1000
                );
            }
        })
                //加水印
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        //设置最大乱序时间
                        .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        //设置和水印的比较值
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        })
                )
                //筛选元素
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior value) throws Exception {
                        return "pv".equals(value.getBehavior());
                    }
                })
                //对商品id进行keyby
                .keyBy(UserBehavior::getItemId)
                //开窗
                .window(SlidingEventTimeWindows.of(Time.minutes(60),Time.minutes(30)))
                //对窗口内的商品进行统计
                .aggregate(new AggregateFunction<UserBehavior, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(UserBehavior value, Long accumulator) {
                        return accumulator + 1;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                }, new ProcessWindowFunction<Long, HotItem, Long, TimeWindow>() {
                    @Override
                    public void process(Long key,
                                        Context context,
                                        Iterable<Long> elements,
                                        Collector<HotItem> out) throws Exception {
                        //将每一个商品的汇总结果输出

                        Long count = elements.iterator().next();
                        out.collect(new HotItem(
                                key,count,context.window().getEnd()
                        ));

                    }
                })

                //处理到这一步的时候遇到一个问题，后面的算子怎么知道我前面窗口的数据有没有到齐，然后再计算topN
                //这时候我们就需要设置一个定时器来进行触发计算，只有当水印时间


                //再对数据进行keyby操作
                //这次按照时间进行分组

                .keyBy(HotItem::getWindowEndTime)
                .process(new KeyedProcessFunction<Long, HotItem, String>() {
                    private ListState<HotItem> listState;

                    //获取状态
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        listState = getRuntimeContext().getListState(new ListStateDescriptor<HotItem>("listState", HotItem.class));
                    }

                    @Override
                    public void processElement(HotItem value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        //当第一个数据进场的时候就触发定时器
                        Iterable<HotItem> hotItems = listState.get();
                        if (!hotItems.iterator().hasNext()) {
                            //注册定时器
                            ctx.timerService().registerEventTimeTimer(value.getWindowEndTime()+2000);
                        }
                        //将结果保存到状态中
                        listState.add(value);
                    }

                    //触发定时任务
                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<String> out) throws Exception {
                        //这个时候开始计算，求top3的数据
                        List<HotItem> hotItems = ITtoListUtil.toList(listState.get());

                        //进行排序
                        hotItems.sort((o1, o2) -> o2.getCount().compareTo(o1.getCount()));
                        StringBuilder msg= new StringBuilder("----------------\n");
                        for (int i = 0; i < Math.min(3,hotItems.size()); i++) {
                            //将结果保存输出
                            msg.append(hotItems.get(i)+"\n");
                        }
                        out.collect(msg.toString());
                    }
                }).print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
